package com.thp.bigdata.rjon;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;


public class RJoin {
	
	
	static class RJoinMapper extends Mapper<LongWritable, Text, Text, InfoBean> {
		
		// 不要把创建对象行代码放在map方法里面，因为map方法会不断地被调用，而我们只需要创建对象来进行赋值而已
		InfoBean bean = new InfoBean();
		Text outKey = new Text();  // 向外输出的key 必须是联系两张表的那个字段
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] fields = line.split(",");
			
			
			// 需要获取文件名字进行区分是哪个表中的数据
			// 先获取文件的切片
			FileSplit fileSplit = (FileSplit) context.getInputSplit();
			// 获取文件名字
			String fileName = fileSplit.getPath().getName();
			
			String pid = ""; // 这个pid是非常关键的，因为map进行分发数据的时候，我们需要根据这个pid来作为key
			if(fileName.startsWith("order")) { // 订单表中的数据
				pid = fields[2];
			    bean.set(Integer.parseInt(fields[0]), fields[1], pid, Integer.parseInt(fields[3]), "", 0, 0, 0);
			} else { // 产品表中的数据
				pid = fields[0];
				bean.set(0, "", pid, 0, fields[1], Integer.parseInt(fields[2]), Float.parseFloat(fields[3]), 1);
			}
			outKey.set(pid);
			context.write(outKey, bean);
		}
	}
	
	
	// reduce 阶段现在我们只需要将Bean输出打印就可以了
	static class RJoinReducer extends Reducer<Text, InfoBean, InfoBean, NullWritable> {
		@Override
		protected void reduce(Text pid, Iterable<InfoBean> beans,Context context) throws IOException, InterruptedException {
			// 产品的bean
			InfoBean pdBean = new InfoBean();
			// 每个产品可能有多个订单  所以使用ArrayList
			ArrayList<InfoBean> orderBeans = new ArrayList<InfoBean>();
			for(InfoBean bean : beans) {
				if("1".equals(bean.getFlag())) {  // 产品文件中的数据
					// 产品文件中每个产品只有一个
					// 注意不能直接使用  pdBean = bean ; 因为都是引用类型
					try {
						BeanUtils.copyProperties(pdBean, bean);
					} catch (IllegalAccessException | InvocationTargetException e) {
						e.printStackTrace();
					}
				} else {  // 订单文件中的数据
					InfoBean odBean = new InfoBean();  // 订单Bean
					try {
						BeanUtils.copyProperties(odBean, bean);
						orderBeans.add(odBean);
					} catch (IllegalAccessException | InvocationTargetException e) {
						e.printStackTrace();
					}
				}
			}
			
			
			// 将两张表中的数据进行拼接
			for(InfoBean bean: orderBeans) {  // 需要将每一个订单中的数据全部都写出去
				bean.setPname(bean.getPname());
				bean.setCategory_id(bean.getCategory_id());
				bean.setPrice(bean.getPrice());
				context.write(bean, NullWritable.get());
			}
			
		}
	}
	
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration conf = new Configuration();
		
		System.err.println(args[0]);
		System.err.println(args[1]);
		
		String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		if(otherArgs.length != 2) {
			System.err.println("Usage : wordcount <in> <out>");
			System.exit(2);
		}
		
		Job job = Job.getInstance(conf);
		
		
		job.setJar("f:/rjoin.jar");
		
		// 指定本业务job要使用的mapper/Reduce业务类
		job.setMapperClass(RJoinMapper.class);
		job.setReducerClass(RJoinReducer.class);
		
		// 指定mapper输出数据的kv类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(InfoBean.class);
		
		// 指定最终输出的数据的kv类型
		job.setOutputKeyClass(InfoBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		
		
		Path path = new Path(otherArgs[1]);
		FileSystem fileSystem = path.getFileSystem(conf);   // 根据path找到这个文件
		if(fileSystem.exists(path)) {
			fileSystem.delete(path, true);  // true的意思是，就算output有东西，也一带删除
		}
		
		
		// 指定job的输入原始文件所在的目录
		// 待处理文件可以在多个目录里面
		FileInputFormat.setInputPaths(job, new Path(args[0]));
		
		// 指定job的输出结果
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		
		// 将job中配置的相关参数,以及job所用的的java类所在的jar包,提交给yarn去运行
		/*job.submit();*/
		boolean res = job.waitForCompletion(true); // 会等待程序处理完成之后,程序才退出
		System.exit(res ? 0 : 1);
	}
	
}
